package com.hkbigdata.ontimer;

import com.hkbigdata.entry.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @author liuanbo
 * @creat 2023-05-17-19:09
 * @see 2194550857@qq.com
 * ws_1001,1684121131000,3
 * ws_1001,1684121140000,1
 * ws_1001,1684121139000,2
 * ws_1001,1684121142000,4
 * ws_1001,1684121131000,3
 * ws_1001,1684121147000,5
 * ws_1001,1684121148000,6
 * <p>
 * 1.测试数据，只输入一条，数据连续10秒没有下载
 * 2.连续输入增加数据，10秒之后定时器触发，走测输出流，防止数据连续增加，需要重置初始化时间
 * 3.数据在10秒之内有减少，删除定时器，但是为了防止删除之后数据再次增加，需要再次重置初始化时间
 */
public class Flink03_Process_VcInrc {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("hadoop102", 9999).map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                String[] split = value.split(",");

                return new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2]));

            }
        });

        //2.分组
        SingleOutputStreamOperator<WaterSensor> result = waterSensorSingleOutputStreamOperator.keyBy(WaterSensor::getId)
                .process(new KeyedProcessFunction<String, WaterSensor, WaterSensor>() {
                    //用来记录上一次传感器数值
                    private Integer lastVc = Integer.MIN_VALUE;
                    //定时器的时间戳
                    private Long TimerTs = Long.MIN_VALUE;

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                        //获取本次传感器的数据
                        Integer vc = value.getVc();
                        //表示第一次来的数据并且传感器的值大于上一次的传感器数值
                        if (vc > lastVc && TimerTs == Long.MIN_VALUE) {
                            long time = ctx.timerService().currentProcessingTime();
                            long ts = time + 10000L;
                            System.out.println("定时器注册时间:" + ts);
                            //注册一个10秒钟定时器
                            ctx.timerService().registerProcessingTimeTimer(ts);

                            //如果在10秒钟之内有数据的下降，那么需要把该条数据定时器删除
                            TimerTs = ts;

                        } else if (vc < lastVc) {//表示有数据的下降
                            ctx.timerService().deleteProcessingTimeTimer(TimerTs);
                            System.out.println("删除定时器" + TimerTs);

                            //如果删除定时器之后数据增长，那么需要把TimerTs重置
                            TimerTs = Long.MIN_VALUE;
                        }
                        //表示不是第一次数据来，并且增长，那么需要将lastVc进行替换
                        lastVc = vc;

                        //输出数据
                        out.collect(value);

                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<WaterSensor> out) throws Exception {
                        ctx.output(new OutputTag<String>("side") {
                        }, "连续10秒数据连续增长");

                        //重置时间戳，目的防止后面的数据持续增长
                        TimerTs = Long.MIN_VALUE;
                    }
                });

        result.print("主流");

        result.getSideOutput(new OutputTag<String>("side") {
        }).print("测输出流");

        env.execute();

    }
}
